package com.crazyice.lee.jobs;

import com.alibaba.fastjson.JSONObject;
import com.crazyice.lee.data.Student;
import com.crazyice.lee.writer.JdbcWriter;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

@Slf4j
public class Kafka2MysqlJob {
    private static Properties properties;

    private static void init() {
        properties = new Properties();
        try (InputStream inputStream = Kafka2MysqlJob.class.getResourceAsStream("/kafka.properties")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        init();

        //设置环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取数据-数据源
        DataStreamSource<String> students = env.addSource(new FlinkKafkaConsumer010("student", new SimpleStringSchema(), properties));
        students.name("从kafka读取数据");
        students.print();

        //处理数据过程，将json string转换为bean
        DataStream<Student> subStudents = students
                .map(key -> {
                    Student student = JSONObject.parseObject(key, Student.class);
                    return student;
                }).name("JSONString转换为Student POJO")
                .flatMap(new FlatMapFunction<Student, Student>() {
                    @Override
                    public void flatMap(Student student, Collector<Student> collector) throws Exception {
                        if (student.getId() % 2 == 0) {
                            collector.collect(student);
                        }
                    }
                }).name("学号为偶数的学生")
                .filter(e -> e.isSex()).name("过滤男生")
                .filter(e -> e.getAge() > 12).name("过滤12岁以上")
                .keyBy(new KeySelector<Student, Integer>() {
                    @Override
                    public Integer getKey(Student student) throws Exception {
                        return student.getAge();
                    }
                }).sum("age").name("年龄求和");

        //写入数据-处理结果
        subStudents.addSink(new JdbcWriter()).name("写入备份学生库");
        env.execute("读取kafka，数据过滤，写入Mysql");
    }
}
